package com.glodon.paas.util.asyn;

import java.lang.reflect.Method;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.glodon.paas.exception.PaasServerExceptionFactory;
import com.glodon.paas.util.ObjectUtil;

/**
 * 异步工作实现类<br>
 * 并发线程数量：最小=1,最大20<br>
 * 任务队列长度：最小=工作线程数,最大100000<br>
 * spring配置如
 * 
 * <bean id="asynWorker" class="com.glodon.paas.util.asyn.AsynWorkerExecutorImpl" init-method="start">
 *      <property name="threadSize" value="1"/>
 *      <property name="taskQueueSize" value="10000"/>
 * </bean>
 * 
 * @author a
 */
public class AsynWorkerExecutorImpl implements AsynWorker {
    Logger LOGGER = LoggerFactory.getLogger(AsynWorkerExecutorImpl.class);
    Logger INVOKE_ERROR_LOGGER = LoggerFactory.getLogger("system.asynWorker.invoke.error.log");
    //Logger INVOKE_ERROR_LOGGER = LoggerFactory.getLogger(AsynWorkerExecutorImpl.class);
    private static final int MAX_THREAD_SIZS = 20;
    private static final int MAX_QUEUE_SIZE = 100000;

    private String preThreadName = "AsynWorkerExecutorImpl"; //线程名前缀
    private int threadSize;                             //线程数 
    private int taskQueueSize;                          //任务队列大小，不能小于线程数
    private boolean closing = true;                     //开始关闭，不接受新任务
    private boolean running = false;                    //线程结束
    private int waitSecond4PollQueue = 120;             //获取任务超时时间秒
    private LinkedBlockingQueue<Runnable> taskQueue;    //任务队列  

    public  AsynWorkerExecutorImpl( ){
    }
    public AsynWorkerExecutorImpl(int nThreads,int queueSize){
         this.threadSize=nThreads;
         this.taskQueueSize=queueSize;
    }
    public void setWaitSecond4PollQueue(int second){
        if(second<1){
            second = 10;
        }
        if(second > 3600){
            second = 3600;
        }
        waitSecond4PollQueue = second;
    }
    public void start(){
        if(threadSize<1){
            threadSize = 1;
        }
        if(threadSize > MAX_THREAD_SIZS){
            threadSize = MAX_THREAD_SIZS;
        }
        if(taskQueueSize < threadSize){
            taskQueueSize = threadSize;
        }
        if(taskQueueSize > MAX_QUEUE_SIZE){
            taskQueueSize = MAX_QUEUE_SIZE;
        }
        
        running = true;
        closing=false;
        taskQueue = new LinkedBlockingQueue<Runnable>(taskQueueSize);       
        // 创建nThreads个线程，并启动
        for(int i=0; i<threadSize; i++){
            Thread thread = new Thread(){
                Runnable task;
                public void run() {
                    while(running){
                        try{
                            //task = queue.take();
                            task = taskQueue.poll(waitSecond4PollQueue, TimeUnit.SECONDS);
                            if(task != null){
                                task.run();
                            }else{
                                LOGGER.debug("get task: null");
                            }                                               
                        } catch (InterruptedException e) {
                            LOGGER.error(e.getMessage(), e);
                        }
                    }                   
                }
            };
            thread.setName(preThreadName + "_" + i);
            thread.setDaemon(true);
            thread.start();
        }
        LOGGER.info("start()");
    }

    public void submit(Runnable task) {
        if(closing){
            //TODO
            throw PaasServerExceptionFactory.systemException("AsynWorker is closing", null);
        }
        
        try {
            taskQueue.put(task);
        } catch (InterruptedException e) { 
            LOGGER.error(e.getLocalizedMessage(),e);
            throw PaasServerExceptionFactory.systemException(e.getMessage(), e);
        }
        if(LOGGER.isDebugEnabled()){
            LOGGER.debug("\r\n---->task submited:\r\n" + task);
        }
    }
    
    /**
     * 异步运行obj对象的methodName方法<br>
     * 推荐使用确定的paramTypes， 当paramTypes为null时系统会根据methodName获取方法，此时需要方法唯一，否则异常<br>
     * 
     *@param params methodName方法需要的参数数组
     *@param paramTypes methodName方法对应的参数类型数组
     */
    @Override
    public void submit(final Object obj, String methodName,
            final Object[] params, Class<?>[] paramTypes) {
        /*
         * 如果参数中带有基本类型的数据，此时反射得到的paramTypes不可用，所以注释掉 // 获取参数类型列表 if(paramTypes
         * == null){ LOGGER.warn("paramTypes is null"); paramTypes = new
         * Class<?>[params.length]; for(int i=0; i< params.length; i++){
         * if(params[i] == null){ //hasParameterNull = true; paramTypes = null;
         * break; } paramTypes[i] = params[i].getClass(); } }
         */

            Method method = null;
            if (paramTypes != null) {
                // 反射获取方法
                try {
                    method = obj.getClass().getMethod(methodName,(Class<?>[]) paramTypes);
                } catch (Exception e) {
                    // TODO: handle exception
                }               
            } else {
                // 通过方法名反射获取方法列表
                Method[] methods = obj.getClass().getMethods();
                int paramLength = 0;
                if (params != null) {
                    paramLength = params.length;
                }
                for (Method m : methods) {
                    if (methodName.equals(m.getName())
                            && m.getParameterTypes().length == paramLength) {
                        if (method != null) {
                            throw PaasServerExceptionFactory.notFoundException("multiple method:" + methodName, null);
                        }
                        method = m;
                    }
                }
            }
            if (method == null) {
                throw PaasServerExceptionFactory.notFoundException("method:" + methodName + " with param:"+ ObjectUtil.toJson(params), null);
            }

            // task需要final类型的参数
            final Method finalMethod = method;
            Runnable task = new Runnable() {
                @Override
                public void run() {
                    try {
                        finalMethod.invoke(obj, params);
                    } catch (Exception e) {
                        INVOKE_ERROR_LOGGER.error("invoke error:"+e.getMessage(), e);
                        INVOKE_ERROR_LOGGER.error("\r\n---->invoke error:\r\n" + invokeLog(obj, finalMethod, params));
                    }
                }

                @Override
                public String toString() {
                    return invokeLog(obj, finalMethod, params);
                }
            };
        submit(task);
    }

    protected String invokeLog(Object obj, Method finalMethod, Object[] params) {
        StringBuilder sb = new StringBuilder();
        sb.append("{").append("\r\n");
        sb.append("method: ").append(finalMethod).append("\r\n");
        sb.append("param: ");
        int len = params.length;
        for(int i=0;i<len;i++){
            sb.append(params[i]);
            if(i < len-1){
                sb.append(",");
            }else{
                sb.append("\r\n");
            }
        }
        sb.append("}").append("\r\n");
        return sb.toString();
    }
    
    @Override
    public void shutdown() {
        closing = true;
        while(taskQueue.size() > 0){
            ObjectUtil.sleep(10, true);
        }
        running = false;
        taskQueue = null;
        //TODO 判断线程运行结束
    }
    
    public void setThreadSize(int threadSize) {
        this.threadSize = threadSize;
    }
    public void setTaskQueueSize(int taskQueueSize) {
        this.taskQueueSize = taskQueueSize;
    }
}
